package skewreduce.framework.physical;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.mapreduce.Job;

import skewreduce.framework.EaggDriver;
import skewreduce.framework.EaggPlan;
import skewreduce.framework.SchedulerEvent;
import skewreduce.framework.logical.LSampleOp;
import skewreduce.lib.Cube2;
import skewreduce.lib.Partition;


public class UniformSample extends PSampleOp {
    private static Log LOG = LogFactory.getLog("PLAN");

    private int initialPartition;

	@Override
	protected Job createJob(Configuration conf) throws IOException {
		throw new UnsupportedOperationException("Uniform sampling does not require a MapReduce job");
	}

	@Override
	public void setup(EaggPlan plan) throws IOException {
        if ( getBound().equals(plan.getPartitionInfo()) ) {
            initialPartition = plan.getDriver().getConf().getInt("skewreduce.partition.initialpartitions",0);
            LOG.info("Force setting the number of initial partitions to "+initialPartition);
        }
    }
	

	@Override
	public SchedulerEvent call() throws Exception {
        SchedulerEvent result = null;

        setup();

        try {
            int numPartitions = initialPartition;
            if ( numPartitions == 0 ) {
                LOG.info("Overriding the number of partitions");
                FileSystem fs = getPlan().getDriver().getFileSystem();
                FileStatus stat = fs.getFileStatus(getInputPath());
                long inputSize = stat.getLen();
                long blockSize = stat.getBlockSize();

                numPartitions = (int)(inputSize / (double)blockSize);
                if ( numPartitions < 2 ) numPartitions = 2;
            }
            
            int numSplits = 0;
            while ( numPartitions > 1 ) {
                numPartitions >>= 1;
                ++numSplits;
            }
            
            Partition.DepthCondition condition = new Partition.DepthCondition(getBound(),numSplits);

            getBound().uniformSplit(condition);

            setStatus(Status.COMPLETE);
            result = new SchedulerEvent.Completion(this);
        } catch ( IOException x ) {
            setStatus(Status.FAILED);
            result = new SchedulerEvent.Error(this,x);
        }
        return result;
	}
	
	public static LSampleOp createLogicalOp(EaggDriver driver) {
		return new LSampleOp(driver) {
			@Override
			public PhysicalOp createInstance() {
				return new UniformSample();
			}
		};
	}
}
